package com.gxlevi.db.util

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010

object FlinkUtil {

  //初始化flink的流式环境
  def initFlinkEnv(): StreamExecutionEnvironment = {
    //Flink流式环境的创建
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置env的处理时间为EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //设置并行度
    env.setParallelism(1)
    //设置checkpoint,开启checkpoint间隔时间为5s
    env.enableCheckpointing(5000)
    //设置处理模式,默认为EXACTLY_ONCE
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    //设置2次checkpoint的间隔
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000)
    //设置超市时长
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    //设置并行度
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    //当程序关闭时触发额外的checkpoint
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    //设置检查点在hdfs中存储的位置
    env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink-checkpoint"))

    env
  }

  def initKafkaFlink(): FlinkKafkaConsumer010[String] = {
    //整合kafka
    val props: Properties = new Properties()
    //kafka集群地址
    props.setProperty("bootstrap.servers", GlobalConfigUtil.bootstrapServers)
    //消费组的id
    props.setProperty("group.id", GlobalConfigUtil.groupId)
    //自动提交拉取到消费端的消息offset到kafka
    props.setProperty("enable.auto.commit", GlobalConfigUtil.enableAutoCommit)
    //自动提交offset到zookeeper的时间间隔单位(ms)
    props.setProperty("auto.commit.interval.ms", GlobalConfigUtil.autoCommitIntervalMs)
    //每次消费最新的数据
    props.setProperty("auto.offset.reset", GlobalConfigUtil.autoOffsetReset)

    val consumer: FlinkKafkaConsumer010[String] = new FlinkKafkaConsumer010[String](
      GlobalConfigUtil.inputTopic,
      new SimpleStringSchema(),
      props
    )

    consumer
  }
}
